package net.bwie.realtime.jtp.dim.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;

import java.util.HashSet;
import java.util.Set;

public class DimDataFilterFunction extends RichFilterFunction<String> {

    private final Set<String> dimSet;

    private final Set<String> operateTypeSet = new HashSet<String>(){{
        add("insert"); add("update"); add("delete");
    }};

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    public DimDataFilterFunction(Set<String> dimSet) {
        this.dimSet = dimSet ;
    }

    @Override
    public boolean filter(String value) throws Exception {
        /*
        {
            "db_name": "jtp_mall",
            "operate_data": {
                "birthday": "2003-01-02",
                "create_time": "2024-06-02 11:18:25.0",
                "login_name": "wcgx7g",
                "nick_name": "倩婷",
                "name": "朱倩婷",
                "user_level": "1",
                "phone_num": "13797988267",
                "id": 174,
                "email": "wcgx7g@163.net",
                "operate_time": "2024-06-02 11:19:14.0"
            },
            "opValue": "unknown",
            "table_name": "user_info",
            "ts_ms": 1748850648577
        }
         */
        JSONObject jsonObject = JSON.parseObject(value);
        String operateType = jsonObject.getString("operate_type");
        String tableName = jsonObject.getString("table_name");

        return (operateTypeSet.contains(operateType) && dimSet.contains(tableName));
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
